//
// Created by LiuYou on 2021/8/18.
//

#include <cstdlib>
#include <netinet/in.h>
#include <cstdio>
#include <cstring>
#include <arpa/inet.h>
#include <cassert>
#include <memory>
#include <sys/epoll.h>
#include <fcntl.h>
#include <csignal>
#include <sys/mman.h>
#include <unistd.h>
#include <sys/wait.h>

/*
 * The file is too complex to perform the context-sensitive data-flow analysis.
 * 该文件太复杂，无法执行上下文相关的数据流分析。
 * 提示慢的原因找到了。
 */


/**
 * @brief 处理一个客户连接必要的数据。
 */
class ClientData {
public:
    ClientData(const sockaddr_in& address, int connectFd, pid_t pid, int* pipeFd) :
            address( address ), connectFd( connectFd ), pid( pid ), pipeFd() {}

    ClientData() :
            address(), connectFd(), pid(), pipeFd() {}

    ~ClientData() = default;

public:
    // 客户端的socket地址
    sockaddr_in address;
    // connectFd文件描述符
    int connectFd;
    // 处理这个连接的子进程的PID
    pid_t pid;
    // 和父进程通信用的管道
    int pipeFd[2];
};


static const int USER_LIMIT = 5;
//static const int PROCESS_LIMIT = 65536;
static const int PROCESS_LIMIT = 300000;
static int userCount;
static std::array<std::unique_ptr<ClientData>, USER_LIMIT + 1> users;
static std::array<std::unique_ptr<int>, PROCESS_LIMIT> subProcess;
static const int MAX_EVENT_NUMBER = 1024;
static int signalPipeFd[2];
static const char* SHM_NAME = "/my_shm";
static int shmFd;
static const int BUFFER_SIZE = 1024;
static char* shareMemory;
static bool stopChild = false;
static int listenFd;
static int epollFd;


static int setFdNonblocking(int fd) {
    int oldOption = fcntl( fd, F_GETFL );
    // 疑惑解开, 将文件描述符设置为非阻塞, 使用的是宏 O_NONBLOCK。
    int newOption = oldOption | O_NONBLOCK;
    fcntl( fd, F_SETFL, newOption );
    return oldOption;
}

static void addFdToEpoll(int epollFd, int fd) {
    epoll_event epollEvent{};
    epollEvent.data.fd = fd;
    epollEvent.events = EPOLLIN | EPOLLET;
    setFdNonblocking( fd );
    epoll_ctl( epollFd, EPOLL_CTL_ADD, fd, &epollEvent );
}

static void addSignal(int sig, void(* signalHandler)(int), bool restart = true) {
    struct sigaction sigAction;
    sigAction.sa_handler = signalHandler;
    if (restart) {
        sigAction.sa_flags |= SA_RESTART;
    }
    sigfillset( &sigAction.sa_mask );
    sigaction( sig, &sigAction, nullptr );
    // 判断。
}

static void signalHandler(int sig) {
    int saveErrno = errno;
    send( signalPipeFd[1], (char*) &sig, 1, 0 );
    errno = saveErrno;
}

static void childTermHandler(int sig) {
    stopChild = true;
}

static void runChildProcess(int id,
                            std::array<std::unique_ptr<ClientData>, USER_LIMIT + 1>& users,
                            char* shareMemory) {
    epoll_event epollEvents[MAX_EVENT_NUMBER];
    int childEpollFd = epoll_create( 5 );
    assert( childEpollFd != -1 );
    int connectFd = users[id]->connectFd;
    addFdToEpoll( childEpollFd, connectFd );

    int pipeFd = users[id]->pipeFd[1];
    addFdToEpoll( childEpollFd, pipeFd );

    addSignal( SIGTERM, childTermHandler, false );

    while (!stopChild) {
        int ready = epoll_wait( childEpollFd, epollEvents, MAX_EVENT_NUMBER, -1 );
        if ((ready < 0) && (errno != EINTR)) {
            printf( "epoll failure\n" );
            break;
        }

        for (int i = 0; i < ready; ++i) {
            int fd = epollEvents[i].data.fd;
            if ((fd == connectFd) && (epollEvents[i].events & EPOLLIN)) {
                memset( shareMemory + id * BUFFER_SIZE, '\0', BUFFER_SIZE );
                ssize_t rcvBytes = recv( connectFd, shareMemory + id * BUFFER_SIZE, BUFFER_SIZE - 1, 0 );
                if (rcvBytes < 0) {
                    if (errno != EAGAIN) {
                        stopChild = true;
                    }
                } else if (rcvBytes == 0) {
                    stopChild = true;
                } else {
                    send( pipeFd, (char*) &id, sizeof( id ), 0 );
                }
            } else if ((fd == pipeFd) && (epollEvents[i].events & EPOLLIN)) {
                int client = 0;
                ssize_t rcvBytes = recv( fd, (char*) &client, sizeof( client ), 0 );
                if (rcvBytes < 0) {
                    if (errno != EAGAIN) {
                        stopChild = true;
                    }
                } else if (rcvBytes == 0) {
                    stopChild = true;
                } else {
                    send( connectFd, shareMemory + client * BUFFER_SIZE, BUFFER_SIZE, 0 );
                }
            } else {
                continue;
            }
        }
    }
    close( connectFd );
    close( pipeFd );
    close( childEpollFd );
}

static void deleteResource() {
    close( signalPipeFd[0] );
    close( signalPipeFd[1] );
    close( listenFd );
    close( epollFd );
    shm_unlink( shareMemory );
}

/**
 * @brief 不找那个BUG了。
 * <p>
 * 现在对在多进程中的全局变量有了新的认识, 不向外扩展的static全局变量在多进程中确实使用较多。
 * @param argc
 * @param argv
 * @return
 */
int main(int argc, char** argv) {
    if (argc < 3) {
        printf( "用法: %s ip port\n", basename( argv[0] ));
        exit( 1 );
    }
    char* ip = argv[1];
    int port = atoi( argv[2] );

    sockaddr_in localAddress{};
    localAddress.sin_family = AF_INET;
    inet_pton( AF_INET, ip, &localAddress.sin_addr );
    localAddress.sin_port = htons( port );

    listenFd = socket( PF_INET, SOCK_STREAM, 0 );
    assert( listenFd >= 0 );

    int optVal = 1;
    int ret = setsockopt( listenFd, SOL_SOCKET, SO_REUSEADDR, &optVal, sizeof( optVal ));
    if (ret == -1) {
        perror( "setsockopt()" );
        exit( 1 );
    }

    ret = bind( listenFd, (sockaddr*) &localAddress, sizeof( localAddress ));
    assert( ret != -1 );

    ret = listen( listenFd, 5 );
    assert( ret != -1 );

    userCount = 0;
    for (int i = 0; i < USER_LIMIT + 1; ++i) {
        users[i] = std::make_unique<ClientData>();
    }
    for (int i = 0; i < PROCESS_LIMIT; ++i) {
        subProcess[i] = std::make_unique<int>( -1 );
    }

    epoll_event epollEvents[MAX_EVENT_NUMBER];
    epollFd = epoll_create( 5 );
    assert( epollFd != -1 );
    addFdToEpoll( epollFd, listenFd );

    ret = socketpair( PF_UNIX, SOCK_STREAM, 0, signalPipeFd );
    assert( ret != -1 );
//    if (ret == -1) {
//        perror( "socketpair()" );
//        exit( 1 );
//    }
    setFdNonblocking( signalPipeFd[1] );
    addFdToEpoll( epollFd, signalPipeFd[0] );

    addSignal( SIGCHLD, signalHandler );
    addSignal( SIGTERM, signalHandler );
    addSignal( SIGINT, signalHandler );
    addSignal( SIGPIPE, SIG_IGN);

    bool stopServer = false;
    bool terminate = false;

    shmFd = shm_open( SHM_NAME, O_CREAT | O_RDWR, 0666 );
    assert( shmFd != -1 );

    ret = ftruncate( shmFd, USER_LIMIT * BUFFER_SIZE );
    assert( ret != -1 );

    shareMemory = (char*) (mmap( nullptr, USER_LIMIT * BUFFER_SIZE,
                                 PROT_READ | PROT_WRITE, MAP_SHARED,
                                 shmFd, 0 ));
    assert( shareMemory != MAP_FAILED );
    close( shmFd );

    while (!stopServer) {
        int ready = epoll_wait( epollFd, epollEvents, MAX_EVENT_NUMBER, -1 );
        if ((ready < 0) && (errno != EINTR)) {
            printf( "epoll failure\n" );
            break;
        }

        for (int i = 0; i < ready; ++i) {
            int fd = epollEvents[i].data.fd;
            if (fd == listenFd) {
                sockaddr_in remoteAddress{};
                socklen_t socklen = sizeof( remoteAddress );
                int connectFd = accept( listenFd, (sockaddr*) &remoteAddress, &socklen );
                if (connectFd < 0) {
                    perror( "accept()" );
                    continue;
                }
                if (userCount >= USER_LIMIT) {
                    const char* info = "too many users\n";
                    printf( "%s", info );
                    send( connectFd, info, strlen( info ), 0 );
                    close( connectFd );
                    continue;
                }
                users[userCount]->address = remoteAddress;
                users[userCount]->connectFd = connectFd;
                ret = socketpair( PF_UNIX, SOCK_STREAM, 0, users[userCount]->pipeFd );
                assert( ret != -1 );
                pid_t pid = fork();
                if (pid < 0) {
                    close( connectFd );
                    continue;
                } else if (pid == 0) {
                    close( epollFd );
                    close( listenFd );
                    close( users[userCount]->pipeFd[0] );
                    close( signalPipeFd[0] );
                    close( signalPipeFd[1] );
                    runChildProcess( userCount, users, shareMemory );
                    munmap((void*) shareMemory, USER_LIMIT * BUFFER_SIZE );
                    exit( 0 );
                } else {
                    close( connectFd );
                    close( users[userCount]->pipeFd[1] );
                    addFdToEpoll( epollFd, users[userCount]->pipeFd[0] );
                    users[userCount]->pid = pid;
                    // todo 错误: 通过调试发现错误在这里。
                    // todo 错误: 通过调试发现 pid 的值为242919。这真的太大了。
                    //  要想运行成功就要增大 PROCESS_LIMIT 的值, 让它的数量级也上至少三十万。
                    // todo 错误: 确实我改为30万后, 运行就成功了。
                    //  这个BUG给我的感觉就是硬件和软件都在在不断的发展。谁能想象到现在PID都排到三十万了呢？
                    *subProcess[pid] = userCount;
                    ++userCount;
                }
            } else if ((fd == signalPipeFd[0]) && (epollEvents[i].events & EPOLLIN)) {
                char signals[1024];
                ssize_t rcvBytes = recv( signalPipeFd[0], signals, sizeof( signals ), 0 );
                if (rcvBytes == -1) {
                    continue;
                } else if (rcvBytes == 0) {
                    continue;
                } else {
                    for (int j = 0; j < rcvBytes; ++j) {
                        switch (signals[j]) {
                            case SIGCHLD: {
                                pid_t pid;
                                int stat;
                                while ((pid = waitpid( -1, &stat, WNOHANG )) > 0) {
                                    int deleteUser = *subProcess[pid];
                                    *subProcess[pid] = -1;
                                    if ((deleteUser < 0) || (deleteUser > USER_LIMIT)) {
                                        continue;
                                    }
                                    epoll_ctl( epollFd, EPOLL_CTL_DEL, users[deleteUser]->pipeFd[0], 0 );
                                    close( users[deleteUser]->pipeFd[0] );

                                    // todo 笔记: 这里报错的原因是unique_ptr不支持拷贝和赋值。
//                                    users[deleteUser] = users[--userCount];
//                                    users[deleteUser].reset( users[--userCount].release());
                                    users[deleteUser] = std::move( users[--userCount] );

                                    *subProcess[users[deleteUser]->pid] = deleteUser;
                                }
                                if (terminate && userCount == 0) {
                                    stopServer = true;
                                }
                                break;
                            }
                            case SIGTERM: {
                                break;
                            }
                            case SIGINT: {
                                printf( "kill all the child now\n" );
                                if (userCount == 0) {
                                    stopServer = true;
                                    break;
                                }
                                for (int k = 0; k < userCount; ++k) {
                                    pid_t pid = users[i]->pid;
                                    kill( pid, SIGTERM );
                                }
                                terminate = true;
                                break;
                            }
                            default: {
                                break;
                            }
                        }
                    }
                }
            } else if (epollEvents[i].events & EPOLLIN) {
                int child = 0;
                ssize_t rcvBytes = recv( fd, (char*) &child, sizeof( child ), 0 );
                printf( "read data from child accross pipe\n" );
                if (rcvBytes == -1) {
                    continue;
                } else if (rcvBytes == 0) {
                    continue;
                } else {
                    for (int j = 0; j < userCount; ++j) {
                        if (users[j]->pipeFd[0] != fd) {
                            printf( "send data to child across pipe\n" );
                            send( users[j]->pipeFd[0], (char*) &child, sizeof( child ), 0 );
                        }
                    }
                }
            }
        }
    }

    deleteResource();

    exit( 0 );
}